package com.ayu.shortlink.engine.schedule;

import com.alibaba.fastjson2.JSON;
import com.ayu.shortlink.engine.common.enums.LocalMessageStatusEnum;
import com.ayu.shortlink.engine.dao.entity.LocalMessageDO;
import com.ayu.shortlink.engine.dao.mapper.LocalMessageMapper;
import com.ayu.shortlink.engine.mq.basics.event.UserGainCouponEvent;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.Objects;

/**
 * 定时扫描本地事务表(补偿)
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class TransactionCompensationTask {
    private final LocalMessageMapper localMessageMapper;
    private final RocketMQTemplate rocketMQTemplate;
    private final Integer RETRY_MAX_TIME = 5;

    // 用XXL-JOB做定时任务比较好 可以避免重复扫描
    @XxlJob(value = "TransactionCompensationTask")
    @Transactional
    public void handler(){
        LambdaQueryWrapper<LocalMessageDO> queryWrapper = new LambdaQueryWrapper<LocalMessageDO>()
                .eq(LocalMessageDO::getStatus, LocalMessageStatusEnum.MESSAGE_PENDING);

        List<LocalMessageDO> localMessageDOS = localMessageMapper.selectList(queryWrapper);

        if (localMessageDOS != null){
            localMessageDOS.forEach(item -> {
                UserGainCouponEvent userGainCouponEvent = JSON.parseObject(item.getMessageBody(), UserGainCouponEvent.class);
                SendResult sendResult = null;
                // 开关 标识是否进入到catch
                boolean flag = false;
                try {
                    sendResult = rocketMQTemplate.syncSend(item.getTopic(), (Message<?>) userGainCouponEvent);
                }catch (Exception e){
                    // 如果超过最大次数的话 那么可能是MQ的问题 日志记录 后期我们去通过日志给用户补偿即可
                    if (Objects.equals(item.getRetryTime(), RETRY_MAX_TIME)){
                        log.error("TransactionCompensationTask执行失败，数据:{}",item);
                        item.setStatus(LocalMessageStatusEnum.MESSAGE_SEND_FAIL.getStatus());
                    }else {
                        item.setRetryTime(item.getRetryTime()+1);
                    }
                    flag = true;
                }
                if (!flag && sendResult != null && sendResult.getSendStatus().equals(SendStatus.SEND_OK)){
                    item.setStatus(LocalMessageStatusEnum.MESSAGE_SEND_OK.getStatus());
                }

                localMessageMapper.updateById(item);
            });
        }
    }
}
